set hive.exec.dynamic.partition = true;
set hive.exec.dynamic.partition.mode = 'nonstrict';
set hive.exec.max.dynamic.partitions = 400;
set hive.exec.max.dynamic.partitions.pernode = 400;

insert overwrite table jms_dm.dm_missing_bind_dt partition (dt='{{ execution_date | cst_ds }}')
select
    t.waybill_no     as billcode               -- 运单号
   ,t.scan_time      as scantime               -- 扫描时间
   ,t.scan_site_code as scansitecode           -- 责任网点code
   ,t.scan_site      as inputsite              -- 责任网点name
   ,dim.agent_code   as subordinate_agent_code -- 代理区code
   ,dim.agent_name   as subordinate_agent_name -- 代理区name
   ,dim.parent_network_code as parent_network_code    -- 加盟商code
   ,dim.parent_network_name as parent_network_name    -- 加盟商name
   ,scan_code --扫描网点code
   ,scan_name --扫描网点name
   ,t.network_type --网点类型
   ,t.first_code as first_code --一段码
   ,warehousing.scan_user as scanuser --入仓最晚扫描时间对应的扫描员
   ,dim.manage_code as manage_region_code --管理大区code
   ,dim.manage_name as manage_region_name --管理大区名称
from (
    select 
        waybill_no
       ,scan_time
       ,scan_site_code
       ,scan_site
       ,scan_code
       ,scan_name
       ,network_type
       ,first_code
    from (
        select 
             *
            ,row_number() over (partition by waybill_no order by send_time desc,duty_time desc) bill_rank
            ,first_code
        from (
            select
                 centersend.waybill_no
                 ,centersend.scan_time
                 ,duty_dept_code.scan_site_code
                 ,duty_dept_code.scan_site
                 ,send.scan_time            as send_time
                 ,duty_dept_code.scan_time  as duty_time
                 ,centersend.scan_site_code as scan_code
                 ,centersend.scan_site      as scan_name
                 ,case when network4.is_distribution_center = 1 then '中心'
                       when network4.is_entrepot = 1 then '集散'
                       else '网点' end as network_type
                 ,waybill.first_code as first_code
            from (
                select *
                from (
                    select *,row_number() over (partition by waybill_no order by scan_time desc) bill_rank
                      from jms_dwd.dwd_tab_barscan_centersend_base_dt
                      where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                        and dt <= '{{ execution_date | cst_ds }}'
                        and date(scan_time) = '{{ execution_date | cst_ds }}') t
                where bill_rank = 1
            ) centersend
            left join(
                select *,row_number() over (partition by waybill_no order by scan_time desc) bill_rank
                from jms_dwd.dwd_tab_barscan_taking_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                 and dt <= '{{ execution_date | cst_ds }}'
            ) taking on centersend.waybill_no = taking.waybill_no
            left join(
                select * 
                from (
                    select waybill_no,row_number() over(partition by waybill_no order by scan_time desc) as rn 
                    from jms_dwd.dwd_tab_barscan_bagging_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                     and dt <= '{{ execution_date | cst_ds }}'
                ) a where a.rn = 1 
            ) bagging on centersend.waybill_no = bagging.waybill_no
            left join jms_dim.dim_lmdm_sys_network_expand network
                on centersend.next_station_code = network.code
            left join jms_dim.dim_lmdm_sys_network_expand taking_network
                on taking.scan_site_code = taking_network.code
            left join jms_dim.dim_lmdm_sys_network network1
                on taking.scan_site_code = network1.code
            left join jms_dim.dim_lmdm_sys_network_expand network2
                on taking.scan_site_code = network2.code
            left join jms_dim.dim_lmdm_sys_network network3
                on network2.in_distribution_code = network3.code
            left join (
                select 
                     waybill_no
                    ,package_receipt_weight
                    ,package_length
                    ,package_wide
                    ,package_high
                    ,waybill_weight
                    ,goods_type_name
                    ,first_code
                from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                 and dt <= '{{ execution_date | cst_ds }}'
            ) waybill on centersend.waybill_no = waybill.waybill_no
            left join (
                select *
                from (
                    select 
                         waybill_no,
                         scan_site_code,
                         row_number() over (partition by waybill_no order by scan_time desc) bill_rank
                    from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                    and dt <= '{{ execution_date | cst_ds }}'
                ) t where bill_rank = 1
            ) arrival on centersend.scan_site_code = arrival.scan_site_code
                     and centersend.waybill_no = arrival.waybill_no
            left join (
                select waybill_no, scan_time
                from (
                    select waybill_no, scan_time
                    from jms_dwd.dwd_tab_barscan_centersend_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                      and dt <= '{{ execution_date | cst_ds }}'
                    union all
                    select waybill_no, scan_time
                    from jms_dwd.dwd_tab_barscan_sitesend_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                      and dt <= '{{ execution_date | cst_ds }}'
                ) t
            ) send on centersend.waybill_no = send.waybill_no
                  and send.scan_time < centersend.scan_time
            left join (
                select waybill_no, scan_site_code, input_site as scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_taking_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, input_site as scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_warehousing_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_sitesend_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_sitearrival_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_centersend_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                 and dt <= '{{ execution_date | cst_ds }}'
            ) duty_dept_code on centersend.waybill_no = duty_dept_code.waybill_no
                              and duty_dept_code.scan_time < centersend.scan_time
            left join jms_dim.dim_lmdm_sys_network network4
                on duty_dept_code.scan_site_code = network4.code
            where substr(centersend.waybill_no, 0, 1) <> 'B'
             and bagging.waybill_no is null
             and network.is_distribution_center = 2
             and (case when network1.is_entrepot <> 1
                       then taking_network.in_distribution_code <> centersend.scan_site_code
                       else 1 = 1 end)
             and (case when network3.is_entrepot = 1
                       then network2.in_distribution_code <> centersend.scan_site_code
                       else 1 = 1 end)
             and (case when centersend.weight is not null then centersend.weight <= 3
                       when centersend.weight is null and waybill.package_receipt_weight is not null
                       then waybill.package_receipt_weight <= 3
                       when centersend.weight is null and waybill.package_receipt_weight is null
                       then waybill.waybill_weight <= 3 end)
             and waybill.package_length <= 30
             and waybill.package_wide <= 30
             and waybill.package_high <= 30
             and arrival.waybill_no is null
             and send.waybill_no is null
             and (waybill.goods_type_name is null or waybill.goods_type_name not in ('生鲜件', '易碎品'))
        ) t
    ) t where bill_rank = 1
) t
left join jms_dim.dim_network_whole_massage dim
   on t.scan_site_code = dim.code
left join (
    select *
    from (
        select 
             waybill_no
              ,scan_site_code
              ,input_site
              ,scan_time
              ,scan_user
              ,row_number() over (partition by waybill_no order by scan_time desc) as rank_num
        from jms_dwd.dwd_tab_barscan_warehousing_base_dt
        where dt >= date_add('{{ execution_date | cst_ds }}', -7)
           and dt <= '{{ execution_date | cst_ds }}'
    ) a where rank_num = 1
) warehousing on t.waybill_no = warehousing.waybill_no
             and t.scan_site_code = warehousing.scan_site_code
where dim.agent_code <> 'Test001'
  and dim.code <> '888888'


union all

select 
      t.waybill_no     as billcode               -- 运单号
    ,t.scan_time      as scantime               -- 扫描时间
    ,t.scan_site_code as scansitecode           -- 责任网点code
    ,t.scan_site      as inputsite              -- 责任网点name
    ,dim.agent_code   as subordinate_agent_code -- 代理区code
    ,dim.agent_name   as subordinate_agent_name -- 代理区name
    ,dim.fran_code    as parent_network_code    -- 加盟商code
    ,dim.fran_name    as parent_network_name    -- 加盟商name
    ,scan_code --扫描网点code
    ,scan_name --扫描网点name
    ,t.network_type --网点类型
    ,t.first_code as first_code --一段码
    ,warehousing.scan_user as scanuser --入仓最晚扫描时间对应的扫描员
    ,dim.manage_code as manage_region_code --管理大区code
    ,dim.manage_name as manage_region_name --管理大区名称
from (    
    select 
         waybill_no
        ,scan_time
        ,scan_site_code
        ,scan_site
        ,scan_code
        ,scan_name
        ,network_type
        ,first_code
    from (
        select 
             *
            ,row_number() over (partition by waybill_no order by send_time desc,duty_time desc) bill_rank
            ,first_code
        from (
            select
                 sitesend.waybill_no
                 ,sitesend.scan_time
                 ,duty_dept_code.scan_site_code
                 ,duty_dept_code.scan_site
                 ,send.scan_time            as send_time
                 ,duty_dept_code.scan_time  as duty_time
                 ,sitesend.scan_site_code   as scan_code
                 ,sitesend.scan_site        as scan_name
                 ,case when network6.is_distribution_center=1 then '中心' when network6.is_entrepot=1 then '集散' else '网点' end as network_type
                 ,waybill.first_code as first_code
            from (
                select *
                from (
                    select *,row_number() over (partition by waybill_no order by scan_time desc) bill_rank
                      from jms_dwd.dwd_tab_barscan_sitesend_base_dt
                      where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                        and dt <= '{{ execution_date | cst_ds }}'
                        and date(scan_time) = '{{ execution_date | cst_ds }}') t
                where bill_rank = 1
            ) sitesend
            left join(
                select *,row_number() over (partition by waybill_no order by scan_time desc) bill_rank
                from jms_dwd.dwd_tab_barscan_taking_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                 and dt <= '{{ execution_date | cst_ds }}'
            ) taking on sitesend.waybill_no = taking.waybill_no
            left join (
                select * 
                from (
                    select waybill_no,row_number() over(partition by waybill_no order by scan_time desc) as rn 
                    from jms_dwd.dwd_tab_barscan_bagging_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                     and dt <= '{{ execution_date | cst_ds }}'
                ) a where a.rn = 1 
            ) bagging on sitesend.waybill_no = bagging.waybill_no
            left join jms_dim.dim_lmdm_sys_network network
                on sitesend.scan_site_code = network.code
            left join jms_dim.dim_lmdm_sys_network network2
                on sitesend.next_station_code = network2.code
            left join jms_dim.dim_lmdm_sys_network_expand network3
                on taking.scan_site_code = network3.code
            left join jms_dim.dim_lmdm_sys_network_expand network4
                on sitesend.scan_site_code = network4.code
            left join jms_dim.dim_lmdm_sys_network_expand network5
                on network3.in_distribution_code = network5.code
            left join (
                select 
                     waybill_no
                    ,package_receipt_weight
                    ,package_length
                    ,package_wide
                    ,package_high
                    ,waybill_weight
                    ,goods_type_name
                    ,first_code
                from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
            ) waybill on sitesend.waybill_no = waybill.waybill_no
                  
            left join (
                select *
                from (
                    select 
                         waybill_no,
                         scan_site_code,
                         row_number() over (partition by waybill_no order by scan_time desc) bill_rank
                    from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                    and dt <= '{{ execution_date | cst_ds }}'
                ) t where bill_rank = 1
            ) arrival on sitesend.scan_site_code = arrival.scan_site_code
                     and sitesend.waybill_no = arrival.waybill_no
            left join (
                select waybill_no, scan_time
                from (
                    select waybill_no, scan_time
                    from jms_dwd.dwd_tab_barscan_centersend_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                      and dt <= '{{ execution_date | cst_ds }}'
                    union all
                    select waybill_no, scan_time
                    from jms_dwd.dwd_tab_barscan_sitesend_base_dt
                    where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                      and dt <= '{{ execution_date | cst_ds }}'
                ) t
            ) send on sitesend.waybill_no = send.waybill_no
                  and send.scan_time < sitesend.scan_time
                  
            left join (
                select waybill_no, scan_site_code, input_site as scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_taking_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, input_site as scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_warehousing_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_sitesend_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_sitearrival_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_centersend_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                  and dt <= '{{ execution_date | cst_ds }}'
                union all
                select waybill_no, scan_site_code, scan_site, scan_time
                from jms_dwd.dwd_tab_barscan_centerarrival_base_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                 and dt <= '{{ execution_date | cst_ds }}'
            ) duty_dept_code on sitesend.waybill_no = duty_dept_code.waybill_no
                              and duty_dept_code.scan_time < sitesend.scan_time
            left join jms_dim.dim_lmdm_sys_network network6
                on duty_dept_code.scan_site_code = network6.code
            where network.is_entrepot = 1
             and substr(sitesend.waybill_no, 0, 1) <> 'B'
             and bagging.waybill_no is null
             and network2.is_distribution_center = 2
             and network2.is_entrepot = 2
             and network3.in_distribution_code <> sitesend.scan_site_code
             and network3.in_distribution_code <> network4.out_distribution_code
             and network5.in_distribution_code <> network4.out_distribution_code
             and (case when sitesend.weight is not null then sitesend.weight <= 3
                       when sitesend.weight is null and waybill.package_receipt_weight is not null then waybill.package_receipt_weight <= 3
                       when sitesend.weight is null and waybill.package_receipt_weight is null then waybill.waybill_weight <= 3
                  end)
             and waybill.package_length <= 30
             and waybill.package_wide <= 30
             and waybill.package_high <= 30
             and arrival.waybill_no is null
             and send.waybill_no is null
             and (waybill.goods_type_name is null or waybill.goods_type_name not in ('生鲜件', '易碎品'))
        ) t
    ) t where bill_rank = 1
) t
left join jms_dim.dim_network_whole_massage dim
   on t.scan_site_code = dim.code
left join (
    select *
    from (
        select 
             waybill_no
              ,scan_site_code
              ,input_site
              ,scan_time
              ,scan_user
              ,row_number() over (partition by waybill_no order by scan_time desc) as rank_num
        from jms_dwd.dwd_tab_barscan_warehousing_base_dt
        where dt >= date_add('{{ execution_date | cst_ds }}', -7)
           and dt <= '{{ execution_date | cst_ds }}'
    ) a where rank_num = 1
) warehousing on t.waybill_no = warehousing.waybill_no
             and t.scan_site_code = warehousing.scan_site_code
where dim.agent_code <> 'Test001'
  and dim.code <> '888888'
DISTRIBUTE BY 1;